package com.huwaiwai.kafka.demo02.producer;

import com.google.common.collect.Lists;
import com.huwaiwai.kafka.KafkaProperties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.List;
import java.util.Properties;

/**
 * 生产者
 * Created by Administrator on 2017/5/1.
 */
public class KafkaProducer02 {

    public static void main(String [] args) throws InterruptedException {
        Properties props = new Properties();
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", KafkaProperties.BROKER_LIST);
        props.put("partitioner.class", "com.huwaiwai.kafka.demo02.partitioner.MyPartitioner");
        props.put("request.required.acks", "1");
        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<>(config);

        for (int i = 0; i <= 1000000; i++) {
            KeyedMessage<String, String> message = new KeyedMessage<>(KafkaProperties.TOPIC, i + "", "Message" + i);
            producer.send(message);
            Thread.sleep(5000);
        }

        List<KeyedMessage<String, String>> messages = Lists.newArrayList();
        for (int i = 0; i <= 1000000; i++) {
            KeyedMessage<String, String> message = new KeyedMessage<>(KafkaProperties.TOPIC, i + "", "Message" + i);
            messages.add(message);
            if (i % 100 == 0) {
                producer.send(messages);
                messages.clear();
            }
        }
        producer.send(messages);
    }
}
